package com.example.batch.task;

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;


/**
 * @author LiYuhang
 * @version 0.1
 * @application
 * @Date 2021/7/17 16:52
 * @Description TODO
 */
@Slf4j
public class LineProcessTasklet implements Tasklet, StepExecutionListener {

    private ArrayBlockingQueue<List<String>> blockingQueue;


    public LineProcessTasklet(ArrayBlockingQueue<List<String>> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        log.info("start ");
    }

    @Override
    public RepeatStatus execute(StepContribution contribution,ChunkContext chunkContext) throws Exception {

        List<String> poll = blockingQueue.poll(1,TimeUnit.SECONDS);

        processDataList(poll);

        log.info(" contribution:{}  chunkContext:{}",contribution,chunkContext);
        return RepeatStatus.FINISHED;
    }

    private void processDataList(List<String> list) throws Exception {
        log.info("list size:{}",list.size());
    }


    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {

        log.info("end ");

        return ExitStatus.COMPLETED;
    }

}
